package com.guandata.cdc.sink;

import com.guandata.cdc.model.sink.AbstractSinkConfig;
import com.guandata.cdc.model.sink.SinkTableNameStrategy;
import com.guandata.cdc.model.sink.sink_table_name_strategy.CustomSinkTableNameStrategy;
import com.guandata.cdc.model.sink.sink_table_name_strategy.PrefixSinkTableNameStrategy;
import lombok.NonNull;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public abstract class GuanDataSinkFactory<T extends AbstractSinkConfig> {
    protected final T config;
    protected final SinkTableNameStrategy sinkTableNameStrategy;

    protected final Schema schema;

    protected GuanDataSinkFactory(@NonNull T config, @NonNull Schema schema) {
        this.config = config;
        this.schema = schema;

        switch (config.sinkTableNameStrategyMode) {
            case PREFIX:
                this.sinkTableNameStrategy = new PrefixSinkTableNameStrategy(config.parameterTool);
                break;
            case CUSTOM:
            default:
                this.sinkTableNameStrategy = new CustomSinkTableNameStrategy(config.parameterTool);
                break;
        }
    }

    public abstract String registerSinkTable(StreamTableEnvironment tableEnv, String tableName);
}
